package mg.spark

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{Row, SparkSession, types}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}


/**
  *
  * @author eureka.wh
  * @since 2019-05-28 17:16
  */
object spark01 extends Logging{

  def main(args: Array[String]): Unit = {

    val spark = SparkSession
                .builder()
                .appName("spark01")
                .master("local[*]")
                .getOrCreate()

    val sc = spark.sparkContext

    val lines = sc.textFile("/Users/yangzong/Desktop/multi_module/mg-nut/visit_source.txt")

//    lines.map(x=>{
//      val tmp = x.split(" ")
//      (getResource(tmp(0)),1)
//    }).reduceByKey(_+_).sortBy(_._2,false).collect().foreach(println)


   val schema = StructType(Array(
      StructField("resource",StringType,nullable = true),
      StructField("num",IntegerType,nullable = true)
    ))


   val rowRDD = lines.map(x=>{
      val tmp = x.split(" ")
      (getResource(tmp(0)),1)
    }).reduceByKey(_+_).sortBy(_._2,false)
      .map(x=>Row(x._1,x._2))

    // rowRDD.map(r=>r.getAs("resource")).foreach(println)

    // rdd ==> dataframe
    val cityDF = spark.createDataFrame(rowRDD,schema)

    val cityDF2 = spark.createDataFrame(rowRDD,schema).filter("num>=5")

    val c = cityDF.join(cityDF2,Seq("resource"),"left_outer")

    c.show(false)

    // dataframe ==> rdd
//    cityDF.rdd.map(x=>{
//      x.getAs[String]("resource")
//    }).foreach(println)


    // 创建全局的表，表名需要global_temp.city
//    cityDF.createOrReplaceGlobalTempView("city")
//    spark.sql("select * from global_temp.city where num != 1").show(false)

    // 非全局表
//    cityDF.createOrReplaceTempView("city")
//    spark.sql("select * from city where num != 1").show(false)

// cityDF.show(false)

      spark.stop()
  }

  def getResource(url:String):String={
    // http://www.wuqiqiop.com/qer/cr/rq/jqo.flv?name=888 47469 193.68.15.92
    var path:String = ""
    val pathTemp = url.replaceFirst("//","")
    var pathIndex = pathTemp.indexOf("/")

    if (pathIndex != -1){
      path = pathTemp.substring(pathIndex)
      pathIndex = path.indexOf("?")
      if (pathIndex != -1){
        path = path.substring(0,pathIndex)
      }
    }
      path
  }

}
